//! 存储引擎模块
//! 
//! 包含内存管理、索引管理、事务管理和持久化功能

pub mod memory;
pub mod index;
pub mod transaction;
pub mod page;
pub mod tuple;
pub mod persistence;
pub mod metadata;
pub mod data_page;
pub mod page_manager_v2;
pub mod index_v2;
pub mod table_v2;
pub mod tuple_format_v2;
pub mod composite_index_v2;
pub mod index_deletion_v2;
pub mod index_rebuild_v2;

#[cfg(test)]
mod transaction_perf_test;

#[cfg(test)]
mod transaction_acid_tests;

#[cfg(test)]
mod transaction_concurrency_tests;

#[cfg(test)]
mod transaction_recovery_tests;

#[cfg(test)]
mod transaction_enhanced_tests;

#[cfg(test)]
mod wal_tests;

use crate::{Error, Result, config::DatabaseConfig};
use crate::sql::{QueryEngine, QueryResult, Value};
use self::persistence::PersistenceManager;
use self::metadata::MetadataManager;
use std::sync::Arc;
use parking_lot::{RwLock, Mutex};
use std::collections::HashMap;
use chrono;

/// 数据库实例
#[derive(Debug)]
pub struct Database {
    /// 数据库配置
    config: DatabaseConfig,
    /// 查询引擎
    query_engine: Arc<QueryEngine>,
    /// 存储引擎
    storage_engine: Arc<RwLock<StorageEngine>>,
    /// 持久化管理器
    persistence_manager: Option<Arc<PersistenceManager>>,
    /// 连接池
    connections: Arc<RwLock<Vec<Arc<Connection>>>>,
}

impl Database {
    /// 创建新的数据库实例
    pub async fn new() -> Result<Self> {
        Self::with_config(DatabaseConfig::default()).await
    }
    
    /// 创建带持久化的数据库实例
    pub async fn new_with_persistence<P: AsRef<std::path::Path>>(data_dir: P) -> Result<Self> {
        let mut config = DatabaseConfig::default();
        config.enable_persistence = true;
        config.data_directory = Some(data_dir.as_ref().to_path_buf());
        Self::with_config(config).await
    }

    /// 使用指定配置创建数据库实例
    pub async fn with_config(config: DatabaseConfig) -> Result<Self> {
        tracing::info!("创建数据库实例，配置: {:?}", config);

        // 创建持久化管理器（如果启用）
        let persistence_manager = if config.enable_persistence {
            if let Some(ref data_dir) = config.data_directory {
                Some(Arc::new(PersistenceManager::new(data_dir).await?))
            } else {
                return Err(Error::config("启用持久化时必须指定数据目录".to_string()));
            }
        } else {
            None
        };

        // 创建存储引擎
        let storage_engine = Arc::new(RwLock::new(
            if let Some(ref persistence) = persistence_manager {
                StorageEngine::with_persistence(&config, persistence.clone())?
            } else {
                StorageEngine::new(&config)?
            }
        ));

        let query_engine = Arc::new(QueryEngine::new(storage_engine.clone()));
        let connections = Arc::new(RwLock::new(Vec::new()));

        Ok(Self {
            config,
            query_engine,
            storage_engine,
            persistence_manager,
            connections,
        })
    }
    
    /// 创建数据库连接
    pub async fn connect(&self) -> Result<Connection> {
        let connection_count = self.connections.read().len();
        
        if connection_count >= self.config.max_connections {
            return Err(Error::connection("连接数已达上限"));
        }
        
        let connection = Connection::new(
            self.query_engine.clone(),
            self.storage_engine.clone(),
        );
        
        self.connections.write().push(Arc::new(connection.clone()));
        
        tracing::info!("创建新连接，当前连接数: {}", connection_count + 1);
        Ok(connection)
    }
    
    /// 获取数据库配置
    pub fn config(&self) -> &DatabaseConfig {
        &self.config
    }

    /// 获取页面管理器
    pub fn get_page_manager(&self) -> Arc<page::PageManager> {
        // 从存储引擎获取页面管理器
        self.storage_engine.read().get_page_manager()
    }
    
    /// 获取连接数
    pub fn connection_count(&self) -> usize {
        self.connections.read().len()
    }
    
    /// 关闭数据库
    pub async fn close(&self) -> Result<()> {
        tracing::info!("关闭数据库");
        
        // 关闭所有连接
        self.connections.write().clear();
        
        // 持久化数据
        let storage = self.storage_engine.read();
        storage.flush().await?;
        
        Ok(())
    }
}

/// 数据库连接
#[derive(Debug, Clone)]
pub struct Connection {
    /// 查询引擎
    query_engine: Arc<QueryEngine>,
    /// 存储引擎
    storage_engine: Arc<RwLock<StorageEngine>>,
    /// 当前事务
    current_transaction: Option<Arc<Transaction>>,
}

impl Connection {
    /// 创建新连接
    fn new(
        query_engine: Arc<QueryEngine>,
        storage_engine: Arc<RwLock<StorageEngine>>,
    ) -> Self {
        Self {
            query_engine,
            storage_engine,
            current_transaction: None,
        }
    }
    
    /// 执行SQL语句
    pub async fn execute(&mut self, sql: &str) -> Result<QueryResult> {
        println!("🔍 Connection开始执行SQL: {}", sql);
        tracing::info!("执行SQL: {}", sql);

        // 创建一个新的QueryEngine实例来执行SQL
        // 这样可以避免Arc<QueryEngine>的可变引用问题
        let storage_engine_clone = Arc::clone(&self.storage_engine);
        let mut query_engine = crate::sql::QueryEngine::new(storage_engine_clone);

        println!("✅ 创建QueryEngine成功，开始执行SQL");
        let result = query_engine.execute(sql)?;
        println!("✅ SQL执行完成，affected_rows: {}", result.get_affected_rows());

        Ok(result)
    }
    
    /// 执行查询
    pub async fn query(&mut self, sql: &str) -> Result<QueryResult> {
        self.execute(sql).await
    }
    
    /// 开始事务
    pub async fn begin_transaction(&mut self) -> Result<Transaction> {
        if self.current_transaction.is_some() {
            return Err(Error::transaction("事务已存在"));
        }

        // 先从事务管理器获取事务ID
        let tx_arc = {
            let storage = self.storage_engine.read();
            storage.transaction_manager.begin_transaction(crate::storage::transaction::IsolationLevel::ReadCommitted, false)?
        };

        let transaction = Transaction::with_id(tx_arc.id, self.storage_engine.clone()).await?;
        self.current_transaction = Some(Arc::new(transaction.clone()));

        tracing::info!("开始事务: {}", transaction.id());
        Ok(transaction)
    }
    
    /// 提交当前事务
    pub async fn commit(&mut self) -> Result<()> {
        if let Some(transaction) = self.current_transaction.take() {
            // 尝试获取事务的独占引用
            match Arc::try_unwrap(transaction) {
                Ok(tx) => {
                    tx.commit().await?;
                    tracing::info!("事务提交成功");
                }
                Err(arc_tx) => {
                    // 如果无法获取独占引用，需要从存储中获取事务Arc
                    let tx_id = arc_tx.id();
                    let storage = self.storage_engine.read();
                    if let Some(tx_arc) = storage.transaction_manager.get_active_transaction(tx_id) {
                        storage.transaction_manager.commit_transaction(tx_arc)?;
                        tracing::info!("事务提交成功");
                    } else {
                        return Err(Error::transaction("事务不存在或已结束"));
                    }
                }
            }
        } else {
            return Err(Error::transaction("没有活动事务"));
        }

        Ok(())
    }
    
    /// 回滚当前事务
    pub async fn rollback(&mut self) -> Result<()> {
        if let Some(transaction) = self.current_transaction.take() {
            match Arc::try_unwrap(transaction) {
                Ok(tx) => {
                    tx.rollback().await?;
                    tracing::info!("事务回滚成功");
                }
                Err(arc_tx) => {
                    // 如果无法获取独占引用，需要从存储中获取事务Arc
                    let tx_id = arc_tx.id();
                    let storage = self.storage_engine.read();
                    if let Some(tx_arc) = storage.transaction_manager.get_active_transaction(tx_id) {
                        storage.transaction_manager.rollback_transaction(tx_arc)?;
                        tracing::info!("事务回滚成功");
                    } else {
                        return Err(Error::transaction("事务不存在或已结束"));
                    }
                }
            }
        } else {
            return Err(Error::transaction("没有活动事务"));
        }

        Ok(())
    }
    
    /// 检查是否在事务中
    pub fn in_transaction(&self) -> bool {
        self.current_transaction.is_some()
    }
}

/// 存储引擎
#[derive(Debug)]
pub struct StorageEngine {
    /// 表存储
    tables: HashMap<String, Table>,
    /// 内存管理器
    memory_manager: memory::MemoryManager,
    /// 索引管理器
    index_manager: index::IndexManager,
    /// 事务管理器
    transaction_manager: transaction::TransactionManager,
    /// 元数据管理器
    metadata_manager: Option<Arc<MetadataManager>>,
}

impl StorageEngine {
    /// 创建新的存储引擎（内存模式）
    pub fn new(config: &DatabaseConfig) -> Result<Self> {
        let page_manager = Arc::new(page::PageManager::new(1000)); // 1000页缓存

        // 在测试环境中使用临时目录
        let wal_path = if cfg!(test) {
            use std::env;
            let temp_dir = env::temp_dir();
            temp_dir.join("cdb_test_wal.log").to_string_lossy().to_string()
        } else {
            // 确保数据目录存在
            std::fs::create_dir_all("./data").unwrap_or(());
            "./data/wal.log".to_string()
        };

        // 创建元数据管理器（即使在内存模式下也需要）
        let metadata_manager = if let Some(ref data_dir) = config.data_directory {
            let metadata_dir = data_dir.join("metadata");
            std::fs::create_dir_all(&metadata_dir).unwrap_or(());
            Some(Arc::new(MetadataManager::new(metadata_dir)?))
        } else {
            // 在内存模式下使用临时目录
            use std::env;
            let temp_dir = env::temp_dir().join("cdb_metadata");
            std::fs::create_dir_all(&temp_dir).unwrap_or(());
            Some(Arc::new(MetadataManager::new(temp_dir)?))
        };

        Ok(Self {
            tables: HashMap::new(),
            memory_manager: memory::MemoryManager::new(config)?,
            index_manager: index::IndexManager::new(),
            transaction_manager: transaction::TransactionManager::new(page_manager, wal_path)?,
            metadata_manager,
        })
    }

    /// 创建带持久化的存储引擎
    pub fn with_persistence(
        config: &DatabaseConfig,
        persistence_manager: Arc<PersistenceManager>
    ) -> Result<Self> {
        // 创建带持久化的页面管理器
        let page_manager = Arc::new(page::PageManager::with_persistence(1000, persistence_manager));

        // 在测试环境中使用临时目录
        let wal_path = if cfg!(test) {
            use std::env;
            let temp_dir = env::temp_dir();
            temp_dir.join("cdb_test_wal.log").to_string_lossy().to_string()
        } else {
            // 使用配置的数据目录
            if let Some(ref data_dir) = config.data_directory {
                std::fs::create_dir_all(data_dir).unwrap_or(());
                data_dir.join("wal.log").to_string_lossy().to_string()
            } else {
                "./data/wal.log".to_string()
            }
        };

        // 创建元数据管理器
        let metadata_manager = if let Some(ref data_dir) = config.data_directory {
            let metadata_dir = data_dir.join("metadata");
            Some(Arc::new(MetadataManager::new(metadata_dir)?))
        } else {
            None
        };

        Ok(Self {
            tables: HashMap::new(),
            memory_manager: memory::MemoryManager::new(config)?,
            index_manager: index::IndexManager::new(),
            transaction_manager: transaction::TransactionManager::new(page_manager, wal_path)?,
            metadata_manager,
        })
    }

    /// 获取页面管理器
    pub fn get_page_manager(&self) -> Arc<page::PageManager> {
        self.transaction_manager.get_page_manager()
    }

    /// 获取元数据管理器
    pub fn get_metadata_manager(&self) -> Option<&Arc<MetadataManager>> {
        self.metadata_manager.as_ref()
    }

    /// 创建表
    pub fn create_table(&mut self, name: String, schema: TableSchema) -> Result<()> {
        if self.tables.contains_key(&name) {
            return Err(Error::query_execution(format!("表 {} 已存在", name)));
        }

        let page_manager = Arc::new(page::PageManager::new(1000));
        let table = Table::new(name.clone(), schema, page_manager);
        self.tables.insert(name, table);

        Ok(())
    }
    
    /// 获取表
    pub fn get_table(&self, name: &str) -> Option<&Table> {
        self.tables.get(name)
    }
    
    /// 获取可变表引用
    pub fn get_table_mut(&mut self, name: &str) -> Option<&mut Table> {
        self.tables.get_mut(name)
    }
    
    /// 删除表
    pub fn drop_table(&mut self, name: &str) -> Result<()> {
        if self.tables.remove(name).is_none() {
            return Err(Error::TableNotFound(name.to_string()));
        }
        
        Ok(())
    }
    
    /// 列出所有表
    pub fn list_tables(&self) -> Vec<String> {
        self.tables.keys().cloned().collect()
    }
    
    /// 刷新数据到持久化存储
    pub async fn flush(&self) -> Result<()> {
        // 简化实现，后续完善
        tracing::debug!("刷新存储引擎数据");
        Ok(())
    }
}

/// 表结构 - 真正的页面式存储实现
#[derive(Debug)]
pub struct Table {
    /// 表名
    name: String,
    /// 表模式
    schema: TableSchema,
    /// 页面管理器 - 核心存储组件
    page_manager: Arc<page::PageManager>,
    /// 数据页面列表 - 存储实际数据的页面ID
    data_pages: Vec<page::PageId>,
    /// 数据页布局管理器 - 管理页面内的元组存储
    data_page_manager: Arc<Mutex<tuple::DataPageManager>>,
    /// 表统计信息
    statistics: TableStatistics,
    /// 当前事务ID（用于MVCC）
    current_transaction_id: Arc<Mutex<u32>>,
}

/// 表行结构 - 支持MVCC的页面式存储
#[derive(Debug, Clone)]
pub struct TableRow {
    /// 行ID - 指向页面中的具体位置
    pub row_id: tuple::RowId,
    /// 行数据 - 从页面中读取的实际数据
    pub values: Vec<Value>,
    /// 创建事务ID
    pub xmin: u32,
    /// 删除事务ID（0表示未删除）
    pub xmax: u32,
}

/// 表统计信息
#[derive(Debug, Clone, Default)]
pub struct TableStatistics {
    /// 总行数
    pub total_rows: u64,
    /// 已删除行数
    pub deleted_rows: u64,
    /// 已更新行数
    pub updated_rows: u64,
    /// 数据页数量
    pub data_pages: u64,
    /// 平均行大小
    pub average_row_size: u32,
    /// 最后更新时间
    pub last_updated: u64,
}

impl Table {
    /// 创建新表 - 真正的页面式存储实现
    pub fn new(name: String, schema: TableSchema, page_manager: Arc<page::PageManager>) -> Self {
        Self {
            name,
            schema,
            page_manager,
            data_pages: Vec::new(),
            data_page_manager: Arc::new(Mutex::new(tuple::DataPageManager::new())),
            statistics: TableStatistics::default(),
            current_transaction_id: Arc::new(Mutex::new(1)),
        }
    }

    /// 获取表名
    pub fn name(&self) -> &str {
        &self.name
    }

    /// 获取表模式
    pub fn schema(&self) -> &TableSchema {
        &self.schema
    }

    /// 获取下一个事务ID
    fn next_transaction_id(&self) -> u32 {
        let mut current_id = self.current_transaction_id.lock();
        *current_id += 1;
        *current_id
    }

    /// 插入行 - 真正的页面式存储实现
    pub fn insert_row(&mut self, values: Vec<Value>) -> Result<tuple::RowId> {
        if values.len() != self.schema.columns.len() {
            return Err(Error::query_execution("行数据列数不匹配"));
        }

        // 执行约束检查
        self.validate_constraints(&values)?;

        // 获取事务ID
        let transaction_id = self.next_transaction_id();

        // 创建元组
        let tuple = tuple::Tuple::from_values(values, transaction_id, 0)?;

        // 获取或分配数据页面
        let page_id = self.get_or_allocate_data_page()?;

        // 在页面中插入元组（简化实现）
        let slot_id = {
            let mut page_manager = self.data_page_manager.lock();
            page_manager.insert_tuple(page_id, &tuple)?
        };

        // 生成行ID
        let row_id = tuple::RowId::new(page_id, slot_id);

        // 更新统计信息
        self.statistics.total_rows += 1;
        self.statistics.last_updated = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_secs();

        tracing::debug!("插入行到表 '{}': {:?}", self.name, row_id);
        Ok(row_id)
    }

    /// 获取或分配数据页面
    /// 如果当前页面有足够空间则使用，否则分配新页面
    fn get_or_allocate_data_page(&mut self) -> Result<page::PageId> {
        // 检查是否有现有页面且有足够空间
        if let Some(&last_page_id) = self.data_pages.last() {
            // 检查页面是否有足够空间（简化实现）
            if let Ok(page) = self.page_manager.read_page(last_page_id) {
                if page.header.free_space > 64 { // 假设需要至少64字节空间
                    return Ok(last_page_id);
                }
            }
        }

        // 分配新的数据页面
        let page_id = self.page_manager.allocate_page(page::PageType::Data)?;
        self.data_pages.push(page_id);

        tracing::debug!("为表 '{}' 分配新数据页面: {}", self.name, page_id);
        Ok(page_id)
    }

    /// 扫描表中的所有可见行（支持MVCC） - 真正的页面式存储实现
    pub fn scan_rows(&self, transaction_id: u32) -> Result<Vec<Vec<Value>>> {
        let mut results = Vec::new();

        // 如果没有数据页面，直接返回空结果
        if self.data_pages.is_empty() {
            tracing::debug!("表 '{}' 没有数据页面，返回空结果", self.name);
            return Ok(results);
        }

        // 遍历所有数据页面
        for &page_id in &self.data_pages {
            // 读取页面
            let page = match self.page_manager.read_page(page_id) {
                Ok(page) => page,
                Err(e) => {
                    tracing::warn!("无法读取页面 {}: {:?}", page_id, e);
                    continue;
                }
            };

            // 从页面中读取所有元组
            let page_manager = self.data_page_manager.lock();
            let tuples = match page_manager.scan_tuples(page_id, &page) {
                Ok(tuples) => tuples,
                Err(e) => {
                    tracing::warn!("无法扫描页面 {} 中的元组: {:?}", page_id, e);
                    continue;
                }
            };

            for tuple in tuples {
                // 检查MVCC可见性
                if self.is_tuple_visible(&tuple, transaction_id) {
                    // 将元组转换为值向量
                    match tuple.to_values(&self.schema) {
                        Ok(values) => {
                            results.push(values);
                        }
                        Err(e) => {
                            tracing::warn!("无法转换元组为值向量: {:?}", e);
                            continue;
                        }
                    }
                }
            }
        }

        tracing::debug!("表 '{}' 扫描完成，找到 {} 行数据", self.name, results.len());
        Ok(results)
    }

    /// 扫描表中的所有可见行并返回RowId（支持MVCC） - 用于UPDATE/DELETE操作
    pub fn scan_rows_with_row_id(&self, transaction_id: u32) -> Result<Vec<(tuple::RowId, Vec<Value>)>> {
        let mut results = Vec::new();

        // 如果没有数据页面，直接返回空结果
        if self.data_pages.is_empty() {
            tracing::debug!("表 '{}' 没有数据页面，返回空结果", self.name);
            return Ok(results);
        }

        // 遍历所有数据页面
        for &page_id in &self.data_pages {
            // 读取页面
            let page = match self.page_manager.read_page(page_id) {
                Ok(page) => page,
                Err(e) => {
                    tracing::warn!("无法读取页面 {}: {:?}", page_id, e);
                    continue;
                }
            };

            // 从页面中读取所有元组
            let page_manager = self.data_page_manager.lock();
            let tuples_with_slots = match page_manager.scan_tuples_with_slot_id(page_id, &page) {
                Ok(tuples) => tuples,
                Err(e) => {
                    tracing::warn!("无法扫描页面 {} 中的元组: {:?}", page_id, e);
                    continue;
                }
            };

            for (slot_id, tuple) in tuples_with_slots {
                // 检查MVCC可见性
                if self.is_tuple_visible(&tuple, transaction_id) {
                    // 将元组转换为值向量
                    match tuple.to_values(&self.schema) {
                        Ok(values) => {
                            let row_id = tuple::RowId::new(page_id, slot_id);
                            results.push((row_id, values));
                        }
                        Err(e) => {
                            tracing::warn!("无法转换元组为值向量: {:?}", e);
                            continue;
                        }
                    }
                }
            }
        }

        tracing::debug!("表 '{}' 扫描完成，找到 {} 行数据", self.name, results.len());
        Ok(results)
    }

    /// 检查元组对指定事务是否可见
    fn is_tuple_visible(&self, tuple: &tuple::Tuple, transaction_id: u32) -> bool {
        // 从元组头部获取事务信息
        let xmin = tuple.header.xmin();
        let xmax = tuple.header.xmax();

        // 如果元组已被删除，且删除事务ID <= 当前事务ID，则不可见
        if xmax > 0 && xmax <= transaction_id {
            return false;
        }

        // 如果元组的创建事务ID <= 当前事务ID，则可见
        xmin <= transaction_id
    }

    /// 更新行（MVCC更新） - 真正的页面式存储实现
    pub fn update_row(&mut self, row_id: tuple::RowId, new_values: Vec<Value>) -> Result<bool> {
        if new_values.len() != self.schema.columns.len() {
            return Err(Error::query_execution("更新数据列数不匹配"));
        }

        // 执行约束检查（对于UPDATE操作，需要特殊处理主键约束）
        self.validate_update_constraints(&new_values, row_id)?;

        let transaction_id = self.next_transaction_id();

        // 读取指定页面
        let page = self.page_manager.read_page(row_id.page_id)?;

        // 从页面中读取元组
        let page_manager = self.data_page_manager.lock();
        if let Some(old_tuple) = page_manager.read_tuple(row_id.page_id, row_id.slot_id, &page)? {
            // 检查元组是否对当前事务可见
            if self.is_tuple_visible(&old_tuple, transaction_id) {
                // 创建新的元组
                let new_tuple = tuple::Tuple::from_values(new_values.clone(), transaction_id, 0)?;

                // 执行MVCC更新：标记旧元组为删除，插入新元组
                let mut old_tuple_updated = old_tuple.clone();
                old_tuple_updated.header.set_xmax(transaction_id);

                // 更新旧元组（标记为删除）
                drop(page_manager); // 释放锁
                let mut page_manager = self.data_page_manager.lock();
                page_manager.update_tuple(row_id.page_id, row_id.slot_id, &old_tuple_updated)?;

                // 插入新元组（简化实现：在同一页面中插入）
                let _new_slot_id = page_manager.insert_tuple(row_id.page_id, &new_tuple)?;

                // 更新统计信息
                self.statistics.updated_rows += 1;
                self.statistics.last_updated = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs();

                tracing::debug!("更新表 '{}' 中的行: {:?}, 新值: {:?}", self.name, row_id, new_values);
                return Ok(true);
            }
        }

        Ok(false)
    }

    /// 删除行（MVCC软删除） - 真正的页面式存储实现
    pub fn delete_row(&mut self, row_id: tuple::RowId) -> Result<bool> {
        let transaction_id = self.next_transaction_id();

        // 读取指定页面
        let page = self.page_manager.read_page(row_id.page_id)?;

        // 从页面中读取元组
        let page_manager = self.data_page_manager.lock();
        if let Some(mut tuple) = page_manager.read_tuple(row_id.page_id, row_id.slot_id, &page)? {
            // 检查元组是否对当前事务可见
            if self.is_tuple_visible(&tuple, transaction_id) {
                // 执行MVCC软删除 - 设置xmax
                tuple.header.set_xmax(transaction_id);

                // 将更新后的元组写回页面
                drop(page_manager); // 释放锁
                let mut page_manager = self.data_page_manager.lock();
                page_manager.update_tuple(row_id.page_id, row_id.slot_id, &tuple)?;

                // 更新统计信息
                self.statistics.deleted_rows += 1;
                self.statistics.last_updated = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs();

                tracing::debug!("删除表 '{}' 中的行: {:?}", self.name, row_id);
                return Ok(true);
            }
        }

        Ok(false)
    }

    /// 获取表统计信息
    pub fn get_statistics(&self) -> &TableStatistics {
        &self.statistics
    }

    /// 获取行数（兼容旧接口）
    pub fn row_count(&self) -> usize {
        self.statistics.total_rows as usize
    }

    /// 获取所有行（兼容旧接口，使用默认事务ID）
    /// 使用真正的页面式存储扫描
    pub fn rows(&self) -> Vec<Vec<Value>> {
        self.scan_rows(u32::MAX).unwrap_or_default()
    }

    /// 验证约束条件
    /// 检查NOT NULL和PRIMARY KEY约束
    fn validate_constraints(&self, values: &[Value]) -> Result<()> {
        // 1. 检查NOT NULL约束
        for (_i, (value, column)) in values.iter().zip(self.schema.columns.iter()).enumerate() {
            if !column.nullable && matches!(value, Value::Null) {
                return Err(Error::query_execution(format!(
                    "列 '{}' 不能为NULL", column.name
                )));
            }
        }

        // 2. 检查PRIMARY KEY约束（唯一性）
        let primary_key_columns: Vec<usize> = self.schema.columns
            .iter()
            .enumerate()
            .filter(|(_, col)| col.primary_key)
            .map(|(i, _)| i)
            .collect();

        if !primary_key_columns.is_empty() {
            // 提取主键值
            let primary_key_values: Vec<&Value> = primary_key_columns
                .iter()
                .map(|&i| &values[i])
                .collect();

            // 检查主键值是否为NULL
            for (pk_col_idx, pk_value) in primary_key_columns.iter().zip(primary_key_values.iter()) {
                if matches!(pk_value, Value::Null) {
                    return Err(Error::query_execution(format!(
                        "主键列 '{}' 不能为NULL",
                        self.schema.columns[*pk_col_idx].name
                    )));
                }
            }

            // 检查主键唯一性
            self.check_primary_key_uniqueness(&primary_key_values, &primary_key_columns)?;
        }

        Ok(())
    }

    /// 检查主键唯一性
    /// 扫描现有数据，确保主键值不重复
    fn check_primary_key_uniqueness(
        &self,
        new_pk_values: &[&Value],
        pk_column_indices: &[usize]
    ) -> Result<()> {
        // 使用最大事务ID来看到所有已提交的数据
        let transaction_id = u32::MAX; // 使用最大事务ID来看到所有数据

        // 扫描所有现有行
        let existing_rows = self.scan_rows(transaction_id)?;

        for existing_row in existing_rows {
            // 提取现有行的主键值
            let existing_pk_values: Vec<&Value> = pk_column_indices
                .iter()
                .map(|&i| &existing_row[i])
                .collect();

            // 比较主键值
            if new_pk_values.len() == existing_pk_values.len() {
                let mut all_equal = true;
                for (new_val, existing_val) in new_pk_values.iter().zip(existing_pk_values.iter()) {
                    if !self.values_equal(new_val, existing_val) {
                        all_equal = false;
                        break;
                    }
                }

                if all_equal {
                    // 构建主键列名列表用于错误消息
                    let pk_column_names: Vec<&str> = pk_column_indices
                        .iter()
                        .map(|&i| self.schema.columns[i].name.as_str())
                        .collect();

                    return Err(Error::query_execution(format!(
                        "主键约束违反：主键 ({}) 的值已存在",
                        pk_column_names.join(", ")
                    )));
                }
            }
        }

        Ok(())
    }

    /// 比较两个值是否相等
    /// 处理不同数据类型的比较逻辑
    fn values_equal(&self, val1: &Value, val2: &Value) -> bool {
        match (val1, val2) {
            (Value::Null, Value::Null) => true,
            (Value::Integer(a), Value::Integer(b)) => a == b,
            (Value::Real(a), Value::Real(b)) => (a - b).abs() < f64::EPSILON,
            (Value::Text(a), Value::Text(b)) => a == b,
            (Value::Boolean(a), Value::Boolean(b)) => a == b,
            // 类型不同或一个为NULL时不相等
            _ => false,
        }
    }

    /// 验证UPDATE操作的约束条件
    /// 与INSERT不同，UPDATE需要排除当前行进行主键唯一性检查
    fn validate_update_constraints(&self, values: &[Value], current_row_id: tuple::RowId) -> Result<()> {
        // 1. 检查NOT NULL约束
        for (i, (value, column)) in values.iter().zip(self.schema.columns.iter()).enumerate() {
            if !column.nullable && matches!(value, Value::Null) {
                return Err(Error::query_execution(format!(
                    "列 '{}' 不能为NULL", column.name
                )));
            }
        }

        // 2. 检查PRIMARY KEY约束（唯一性，排除当前行）
        let primary_key_columns: Vec<usize> = self.schema.columns
            .iter()
            .enumerate()
            .filter(|(_, col)| col.primary_key)
            .map(|(i, _)| i)
            .collect();

        if !primary_key_columns.is_empty() {
            // 提取主键值
            let primary_key_values: Vec<&Value> = primary_key_columns
                .iter()
                .map(|&i| &values[i])
                .collect();

            // 检查主键值是否为NULL
            for (pk_col_idx, pk_value) in primary_key_columns.iter().zip(primary_key_values.iter()) {
                if matches!(pk_value, Value::Null) {
                    return Err(Error::query_execution(format!(
                        "主键列 '{}' 不能为NULL",
                        self.schema.columns[*pk_col_idx].name
                    )));
                }
            }

            // 检查主键唯一性（排除当前行）
            self.check_primary_key_uniqueness_for_update(&primary_key_values, &primary_key_columns, current_row_id)?;
        }

        Ok(())
    }

    /// 检查UPDATE操作的主键唯一性（排除当前行）
    fn check_primary_key_uniqueness_for_update(
        &self,
        new_pk_values: &[&Value],
        pk_column_indices: &[usize],
        current_row_id: tuple::RowId
    ) -> Result<()> {
        // 使用最大事务ID来看到所有已提交的数据
        let transaction_id = u32::MAX;

        // 扫描所有现有行（包含行ID）
        let existing_rows_with_id = self.scan_rows_with_row_id(transaction_id)?;

        for (row_id, existing_row) in existing_rows_with_id {
            // 跳过当前正在更新的行
            if row_id.page_id == current_row_id.page_id && row_id.slot_id == current_row_id.slot_id {
                continue;
            }

            // 提取现有行的主键值
            let existing_pk_values: Vec<&Value> = pk_column_indices
                .iter()
                .map(|&i| &existing_row[i])
                .collect();

            // 比较主键值
            if new_pk_values.len() == existing_pk_values.len() {
                let mut all_equal = true;
                for (new_val, existing_val) in new_pk_values.iter().zip(existing_pk_values.iter()) {
                    if !self.values_equal(new_val, existing_val) {
                        all_equal = false;
                        break;
                    }
                }

                if all_equal {
                    // 构建主键列名列表用于错误消息
                    let pk_column_names: Vec<&str> = pk_column_indices
                        .iter()
                        .map(|&i| self.schema.columns[i].name.as_str())
                        .collect();

                    return Err(Error::query_execution(format!(
                        "主键约束违反：主键 ({}) 的值已存在",
                        pk_column_names.join(", ")
                    )));
                }
            }
        }

        Ok(())
    }
}

/// 表模式
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct TableSchema {
    /// 列定义
    pub columns: Vec<ColumnDefinition>,
}

impl TableSchema {
    /// 创建新的表模式
    pub fn new(columns: Vec<ColumnDefinition>) -> Self {
        Self { columns }
    }
    
    /// 获取列名列表
    pub fn column_names(&self) -> Vec<String> {
        self.columns.iter().map(|col| col.name.clone()).collect()
    }
}

/// 列定义
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ColumnDefinition {
    /// 列名
    pub name: String,
    /// 数据类型
    pub data_type: DataType,
    /// 是否可为空
    pub nullable: bool,
    /// 是否为主键
    pub primary_key: bool,
}

/// 数据类型
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum DataType {
    Integer,
    BigInt,
    Real,
    Double,
    Text(Option<usize>), // 可选长度限制
    Boolean,
    Date,
    Timestamp,
    Decimal(Option<(u8, u8)>), // 精度和小数位数
}

/// 事务
#[derive(Debug, Clone)]
pub struct Transaction {
    /// 事务ID
    id: u64,
    /// 存储引擎引用
    storage_engine: Arc<RwLock<StorageEngine>>,
    /// 事务状态
    status: TransactionStatus,
}

impl Transaction {
    /// 创建新事务
    async fn new(storage_engine: Arc<RwLock<StorageEngine>>) -> Result<Self> {
        let tx_arc = {
            let storage = storage_engine.read();
            storage.transaction_manager.begin_transaction(
                crate::storage::transaction::IsolationLevel::ReadCommitted,
                false
            )?
        };

        Ok(Self {
            id: tx_arc.id,
            storage_engine,
            status: TransactionStatus::Active,
        })
    }

    /// 使用指定ID创建事务
    async fn with_id(id: u64, storage_engine: Arc<RwLock<StorageEngine>>) -> Result<Self> {
        Ok(Self {
            id,
            storage_engine,
            status: TransactionStatus::Active,
        })
    }
    
    /// 获取事务ID
    pub fn id(&self) -> u64 {
        self.id
    }
    
    /// 提交事务
    pub async fn commit(mut self) -> Result<()> {
        if self.status != TransactionStatus::Active {
            return Err(Error::transaction("事务状态无效"));
        }
        
        self.status = TransactionStatus::Committed;
        
        // 实际的提交逻辑
        let storage = self.storage_engine.read();
        if let Some(tx_arc) = storage.transaction_manager.get_active_transaction(self.id) {
            storage.transaction_manager.commit_transaction(tx_arc)?;
        } else {
            return Err(Error::transaction("事务不存在或已结束"));
        }
        
        Ok(())
    }
    
    /// 回滚事务
    pub async fn rollback(mut self) -> Result<()> {
        if self.status != TransactionStatus::Active {
            return Err(Error::transaction("事务状态无效"));
        }
        
        self.status = TransactionStatus::Aborted;
        
        // 实际的回滚逻辑
        let storage = self.storage_engine.read();
        if let Some(tx_arc) = storage.transaction_manager.get_active_transaction(self.id) {
            storage.transaction_manager.rollback_transaction(tx_arc)?;
        } else {
            return Err(Error::transaction("事务不存在或已结束"));
        }
        
        Ok(())
    }
}

/// 事务状态
#[derive(Debug, Clone, PartialEq)]
pub enum TransactionStatus {
    Active,
    Committed,
    Aborted,
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[tokio::test]
    async fn test_database_creation() {
        let db = Database::new().await.unwrap();
        assert_eq!(db.connection_count(), 0);
    }
    
    #[tokio::test]
    async fn test_connection_creation() {
        let db = Database::new().await.unwrap();
        let _conn = db.connect().await.unwrap();
        assert_eq!(db.connection_count(), 1);
    }
    
    #[test]
    fn test_table_creation() {
        let config = DatabaseConfig::default();
        let mut storage = StorageEngine::new(&config).unwrap();

        let schema = TableSchema::new(vec![
            ColumnDefinition {
                name: "id".to_string(),
                data_type: DataType::Integer,
                nullable: false,
                primary_key: true,
            },
            ColumnDefinition {
                name: "name".to_string(),
                data_type: DataType::Text(Some(50)),
                nullable: false,
                primary_key: false,
            },
        ]);

        storage.create_table("users".to_string(), schema).unwrap();
        assert!(storage.get_table("users").is_some());
    }

    #[test]
    fn test_table_mvcc_operations() {
        let config = DatabaseConfig::default();
        let mut storage = StorageEngine::new(&config).unwrap();

        let schema = TableSchema::new(vec![
            ColumnDefinition {
                name: "id".to_string(),
                data_type: DataType::Integer,
                nullable: false,
                primary_key: true,
            },
            ColumnDefinition {
                name: "name".to_string(),
                data_type: DataType::Text(Some(50)),
                nullable: false,
                primary_key: false,
            },
        ]);

        storage.create_table("test_table".to_string(), schema).unwrap();

        let table = storage.get_table_mut("test_table").unwrap();

        // 插入一些测试数据
        let row1 = vec![Value::Integer(1), Value::Text("Alice".to_string())];
        let row2 = vec![Value::Integer(2), Value::Text("Bob".to_string())];

        let row_id1 = table.insert_row(row1).unwrap();
        let row_id2 = table.insert_row(row2).unwrap();

        // 验证插入成功
        assert_eq!(table.row_count(), 2);

        // 测试MVCC扫描 - 使用足够大的事务ID来看到所有数据
        let rows = table.scan_rows(u32::MAX).unwrap();
        assert_eq!(rows.len(), 2);

        // 测试删除
        let deleted = table.delete_row(row_id1).unwrap();
        assert!(deleted);

        // 验证统计信息
        let stats = table.get_statistics();
        assert_eq!(stats.total_rows, 2);
        assert_eq!(stats.deleted_rows, 1);
    }

    #[test]
    fn test_table_serialization_deserialization() {
        let page_manager = Arc::new(page::PageManager::new(10));
        let schema = TableSchema::new(vec![
            ColumnDefinition {
                name: "id".to_string(),
                data_type: DataType::Integer,
                nullable: false,
                primary_key: true,
            },
            ColumnDefinition {
                name: "name".to_string(),
                data_type: DataType::Text(None),
                nullable: true,
                primary_key: false,
            },
            ColumnDefinition {
                name: "score".to_string(),
                data_type: DataType::Real,
                nullable: false,
                primary_key: false,
            },
        ]);

        let mut table = Table::new("test".to_string(), schema, page_manager);

        // 插入包含不同数据类型的行
        let row = vec![
            Value::Integer(42),
            Value::Text("Test User".to_string()),
            Value::Real(95.5),
        ];

        let row_id = table.insert_row(row.clone()).unwrap();

        // 扫描并验证数据 - 使用足够大的事务ID
        let scanned_rows = table.scan_rows(u32::MAX).unwrap();
        assert_eq!(scanned_rows.len(), 1);

        // 验证数据类型和值
        let retrieved_row = &scanned_rows[0];
        assert_eq!(retrieved_row.len(), 3);

        match &retrieved_row[0] {
            Value::Integer(i) => assert_eq!(*i, 42),
            _ => panic!("Expected Integer"),
        }

        match &retrieved_row[1] {
            Value::Text(s) => assert_eq!(s, "Test User"),
            _ => panic!("Expected Text"),
        }

        match &retrieved_row[2] {
            Value::Real(f) => assert!((f - 95.5).abs() < 0.001),
            _ => panic!("Expected Real"),
        }
    }
    
    #[test]
    fn test_table_data_insertion() {
        let page_manager = Arc::new(page::PageManager::new(10));
        let schema = TableSchema::new(vec![
            ColumnDefinition {
                name: "id".to_string(),
                data_type: DataType::Integer,
                nullable: false,
                primary_key: true,
            },
        ]);

        let mut table = Table::new("test".to_string(), schema, page_manager);
        let _row_id = table.insert_row(vec![Value::Integer(1)]).unwrap();

        assert_eq!(table.row_count(), 1);
    }
}
